package collect

import (
	"com.cyt.common/library/kafka"
	"context"
	"fmt"
	"github.com/gogf/gf/frame/g"
	"github.com/gogf/gf/net/ghttp"
	"github.com/gogf/gf/encoding/gjson"
	"github.com/gogf/gf/os/gtime"
	"sync"
)

func CollectInfo() func(r *ghttp.Request) {
	return func(r *ghttp.Request) {
		r.Middleware.Next()
		info, _ := gjson.Encode(g.Map{
			"status": r.Response.Status, //状态码
			"request_time": gtime.NewFromTimeStamp(r.EnterTime), //请求时间
			"response_time": gtime.Datetime(), //响应时间
			"execute_time": float64(gtime.Now().TimestampMilli()-r.EnterTime), //程序执行时间
			"method": r.Method, //请求方式
			"host": r.Host,	//请求地址
			"url": r.URL.Path, //请求地址
			"proto": r.Proto, //请求协议
			"ip": r.GetClientIp(), //请求ip
			"user_agent": r.UserAgent(), //客户端代理
			"trace_id": r.GetCtxVar("X-Trace-Id"), //追踪id
			"appid": r.GetCtxVar("AppId"), //平台id
			"uid": r.GetCtxVar("MemberId"), //用户id
		})
		var messages []map[string]string
		messages = append(messages, map[string]string{"": string(info)})
		KafkaPush("common_collect_metric", messages)
	}
}


func KafkaPush(topic string, messages []map[string]string) {
	client := kafka.Client()
	ctx, cancelable := context.WithCancel(context.Background())
	producer := client.KAsyncProducer(ctx)

	// obtain the background kafka.Producer process, and the error channel to consume
	processor, errors := producer.Background()
	go processor()
	// if not consuming errors in main thread of exec,
	// use wait group (or http.ListenAndServe, etc.) to block
	wg := &sync.WaitGroup{}

	// consume errors channel until kafka.Producer closes it, when context is canceled
	wg.Add(1)
	go func() {
		defer wg.Done()

		// consume this until library closes it for us, indicating client has shut down
		for err := range errors {
			if err != nil {
				fmt.Printf("Kafka message error: %s", err)
			}
		}
	}()

	if err := producer.Send(topic, messages); err != nil {
		fmt.Println(err.Error()) // only happens if context is cancelled while Send() is blocked on full queue
	}

	// signal the producer's background client to gracefully shut down
	cancelable()
	// wait for the client to shut down after draining outgoing message and error queues
	wg.Wait()
}